package main.test.GraphFramesAPI

import org.apache.spark.graphx.Graph
import org.apache.spark.sql.{Encoders, Row, SparkSession}
import org.graphframes.GraphFrame

object GraphxGFConversion {
  def main(args: Array[String]): Unit = {
    //  def constructGraph: GraphFrame = {
    //    The following example demonstrates how
    //    to create a GraphFrame from vertex and edge DataFrame
//    val sparkConf = new SparkConf()
//    sparkConf.setAppName("createGraph").setMaster("local[*]")
//    val sc = new SparkContext(sparkConf)
    val spark = SparkSession.builder()
      .appName(this.getClass.getName)
      .master("local[*]")
      .getOrCreate()

    val v = spark.createDataFrame(List(
      ("a", "Alice", 34),
      ("b", "Bob", 36),
      ("c", "Charlie", 30),
      ("d", "David", 29),
      ("e", "Esther", 32),
      ("f", "Fanny", 36),
      ("g", "Gabby", 60)
    )).toDF("id", "name", "age")

    // Edge DataFrame
    val e = spark.createDataFrame(List(
      ("a", "b", "friend"),
      ("b", "c", "follow"),
      ("c", "b", "follow"),
      ("f", "c", "follow"),
      ("e", "f", "follow"),
      ("e", "d", "friend"),
      ("d", "a", "friend"),
      ("a", "e", "friend")
    )).toDF("src", "dst", "relationship")

    e.show(false)


    // Create a GraphFrame
    val g = GraphFrame(v, e)

    /**
     * We provide utilities for converting between GraphFrame and GraphX graphs. See the
     * GraphX User Guide http://spark.apache.org/docs/latest/graphx-programming-guide.html
     * for details on GraphX
     *
     *
     */

    //GraphFrame to GraphX
    /**
     * Conversion to GraphX creates a GraphX Graph
     * which has Long vertex IDs and attributes of type Row.
     *
     * Vertex and edge attributes are the original
     * rows in vertices and edges, respectively.
     *
     * Note that vertex (and edge) attributes include
     * vertex IDs (and source, destination IDs)
     * in order to support non-Long vertex IDs.
     * If the vertex IDs are not convertible to Long values,
     * then the values are indexed in order to generate
     * corresponding Long vertex IDs (which is an expensive operation).
     *
     * The column ordering of the returned Graph vertex
     * and edge attributes are specified by GraphFrame.
     * vertexColumns and GraphFrame.edgeColumns, respectively.
     */

    // Graphx to GF
    /**
     * GraphFrame provides two conversions methods. The first takes any GraphX graph and converts the vertex and edge RDDs into DataFrames using schema inference. Those DataFrames are then used to create a GraphFrame.
     *
     * The second conversion method is more complex and
     * is useful for users with existing GraphX code.
     * Its main purpose is to support workflows of the following form:
     * (1) convert a GraphFrame to GraphX,
     * (2) run GraphX code to augment the GraphX graph with new vertex or edge attributes, and
     * (3) merge the new attributes back into the original GraphFrame.
     *
     * For example, given:
     *
     * GraphFrame originalGraph
     * GraphX Graph[String, Int] graph with
     * a String vertex attribute we want to
     * call “category” and an Int edge attribute
     * we want to call “count”
     * We can call fromGraphX(originalGraph, graph, Seq("category"), Seq("count"))
     * to produce a new GraphFrame.
     * The new GraphFrame will be an augmented version of originalGraph,
     * with new GraphFrame.vertices
     * column “category” and new GraphFrame.edges column “count” added.
     */

    //Example Conversions
    /**
     * It demonstrates simple GraphFrame-GraphX conversions
     *
     */

    // Convert to GraphX
    val gx: Graph[Row, Row] = g.toGraphX
    g.vertices.printSchema()

//    gx.vertices.foreach(x=>(x._1,



    // Convert back to GraphFrame.
    // Note that the schema is changed because of constraints in the GraphX API.
    // Error!
    //  no Encoder found for org.apache.spark.sql.Row
    //- field (class: "org.apache.spark.sql.Row", name: "_2")
    //- root class: "scala.Tuple2"
    import spark.implicits._
    val g2 = GraphFrame.fromGraphX[Row, Row](
      gx.mapVertices((_, row) => Row(row.getString(1))).mapEdges(e => Row(e.attr.getString(1)))
    )


  }

}
